package com.corx.network.core.network;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.corx.network.core.exception.NetworkException;
import com.corx.network.core.holder.DeviceStateManagerHolder;
import com.corx.network.core.holder.NetworkStateManagerHolder;
import com.corx.network.core.listener.ReceivedMessageListener;
import com.corx.network.core.manager.DeviceStateManager;
import com.corx.network.core.manager.NetworkStateManager;
import com.corx.network.core.model.ReceivedMessage;
import com.corx.network.core.model.SendMessage;
import com.corx.network.core.type.NetworkStateEnum;
import com.corx.network.core.type.NetworkTypeEnum;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author gx
 * @version 1.0.0
 */
@Slf4j
public class MqttNetworkClient implements MqttNetwork {

    private final ObjectMapper objectMapper = new ObjectMapper();
    private final Map<String, ReceivedMessageListener<ReceivedMessage, ReceivedMessage>> topicListeners = new ConcurrentHashMap<>();
    private final Map<String, Integer> subscribedTopics = new ConcurrentHashMap<>();
    private volatile NetworkStateEnum currentState = NetworkStateEnum.OFFLINE;

    private MqttClient client;

    @Getter
    @Setter
    private String clientId;
    @Getter
    @Setter
    private String brokerUrl;
    @Getter
    @Setter
    private String username;
    @Getter
    @Setter
    private String password;
    @Getter
    @Setter
    private Integer keepAlive = 60;
    @Getter
    @Setter
    private Boolean cleanSession = true;
    @Getter
    @Setter
    private Boolean autoConnect = true;

    public MqttNetworkClient() {
        changeNetworkState(NetworkStateEnum.OFFLINE);
    }

    @Override
    public String getKey() {
        return clientId;
    }

    @Override
    public NetworkTypeEnum getType() {
        return NetworkTypeEnum.MQTT_CLIENT;
    }

    @Override
    public void connect() throws NetworkException {
        try {

            if (isConnected()) {
                throw new NetworkException(String.format("MQTT client 已经连接，跳过重复连接：type = %s ，key = %s", getType(), getKey()));
            }
            changeNetworkState(NetworkStateEnum.CONNECTING);

            client = new MqttClient(brokerUrl, clientId);
            client.setCallback(mqttCallbackExtended());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setAutomaticReconnect(autoConnect);
            options.setCleanSession(cleanSession);
            options.setKeepAliveInterval(keepAlive);

            if (username != null) {
                options.setUserName(username);
            }

            if (password != null) {
                options.setPassword(password.toCharArray());
            }

            client.connect(options);
            changeNetworkState(NetworkStateEnum.ONLINE);

        } catch (Exception e) {
            log.error("MQTT client 连接失败：type = {} ，key = {}", getType(), getKey(), e);
            // 通知连接失败
            changeNetworkState(NetworkStateEnum.ERROR);
            throw new NetworkException(String.format("MQTT client 连接失败：type = %s ，key = %s", getType(), getKey()), e);
        }
    }

    @Override
    public void disconnect() {
        changeNetworkState(NetworkStateEnum.OFFLINE);
        try {
            if (isConnected()) {
                client.disconnect();
            }
        } catch (Exception e) {
            log.error("MQTT client 断开连接失败：type = {} ，key = {}", getType(), getKey(), e);
        }
    }

    @Override
    public boolean isConnected() {
        return this.currentState == NetworkStateEnum.ONLINE;
    }

    @Override
    public void sendMessage(SendMessage message) throws NetworkException {
        try {
            byte[] bytesToSend = message.isOnlyPayload()
                ? message.getPayload()
                : objectMapper.writeValueAsBytes(message);

            MqttMessage mqttMsg = new MqttMessage(bytesToSend);
            mqttMsg.setQos(Integer.parseInt(message.getHeaders().getOrDefault("qos", "0")));
            client.publish(message.getTarget(), mqttMsg);
        } catch (Exception e) {
            log.error("MQTT client 发送消息失败：message = {} ，type = {} ，key = {}", message == null ? "null" : message, getType(), getKey(), e);
            throw new NetworkException(String.format("MQTT client 发送消息失败：type = %s ，key = %s", getType(), getKey()), e);
        }
    }


    @Override
    public void subscribe(String topic, int qos, ReceivedMessageListener<ReceivedMessage, ReceivedMessage> listener) throws Exception {
        try {
            topicListeners.put(topic, listener);
            subscribedTopics.put(topic, qos);
            client.subscribe(topic, qos);
        } catch (MqttException e) {
            log.error("MQTT client 订阅失败：topic = {} ，type = {} ，key = {}", topic, getType(), getKey(), e);
            throw new NetworkException(String.format("MQTT client 订阅失败：topic = %s ，type = %s ，key = %s", topic, getType(), getKey()), e);
        }
    }

    @Override
    public void unsubscribe(String topic) throws Exception {
        try {
            topicListeners.remove(topic);
            subscribedTopics.remove(topic);
            client.unsubscribe(topic);
        } catch (MqttException e) {
            log.error("MQTT client 取消订阅失败：topic = {} ，type = {} ，key = {}", topic, getType(), getKey(), e);
            throw new NetworkException(String.format("MQTT client 取消失败：topic = %s ， type = %s ，key = %s", topic, getType(), getKey()), e);
        }
    }

    @Override
    public void destroy() {
        changeNetworkState(NetworkStateEnum.OFFLINE);
        try {
            if (client != null) {
                if (client.isConnected()) {
                    client.disconnect();
                }
                // 释放资源
                client.close();
            }
            topicListeners.clear();
            log.info("MQTT client 资源已释放：type = {} ，key = {}", getType(), getKey());
        } catch (Exception e) {
            log.error("MQTT client 销毁失败：type = {} ，key = {}", getType(), getKey(), e);
        }
    }

    private MqttCallbackExtended mqttCallbackExtended() {
        return new MqttCallbackExtended() {
            @Override
            public void connectComplete(boolean reconnect, String serverURI) {
                log.info("MQTT client 连接{}：serverURI = {} ，type = {} ，key = {}", reconnect ? "重连成功" : "首次连接成功", serverURI, getType(), getKey());
                if (reconnect) {
                    changeNetworkState(NetworkStateEnum.ONLINE);
                    log.info("正在恢复订阅列表...");
                    subscribedTopics.forEach((topic, qos) -> {
                        try {
                            client.subscribe(topic, qos);
                            log.info("已恢复订阅：topic = {} ，qos={} ，type = {} ，key = {}", topic, qos, getType(), getKey());
                        } catch (Exception e) {
                            log.error("恢复订阅失败：topic = {} ，qos={} ，type = {} ，key = {}", topic, qos, getType(), getKey(), e);
                        }
                    });
                }
            }

            @Override
            public void connectionLost(Throwable cause) {
                log.warn("MQTT client 连接丢失：type = {} ，key = {}", getType(), getKey(), cause);
                changeNetworkState(NetworkStateEnum.ERROR);
            }

            @Override
            public void messageArrived(String topic, MqttMessage message) {
                byte[] payload = message.getPayload();
                ReceivedMessageListener<ReceivedMessage, ReceivedMessage> consumers = topicListeners.get(topic);
                if (consumers == null) {
                    log.warn("MQTT client 接收消息没找到topicListeners处理，已经return不进行任何处理，topic = {} ，type = {} ，key = {} ", topic, getType(), getKey());
                    return;
                }
                ReceivedMessage receivedMessage;
                if (isPlatformMessageFormat(payload)) {
                    log.debug("MQTT client 接收的payload是平台标准协议，topic = {} ，type = {} ，key = {} ", topic, getType(), getKey());
                    try {
                        receivedMessage = objectMapper.readValue(payload, ReceivedMessage.class);
                        receivedMessage.setNetworkType(getType());
                        receivedMessage = consumers.apply(receivedMessage);
                    } catch (IOException e) {
                        log.error("MQTT client 解析接收消息异常：转换为平台标准协议ReceivedMessage.class发生错误，保留原始数据发送给执行链条，topic = {} ，type = {} ，key = {}", topic, getType(), getKey(), e);
                        receivedMessage = ReceivedMessage.builder(topic, payload)
                            .build();
                        receivedMessage.setNetworkType(getType());
                        receivedMessage = consumers.apply(receivedMessage);
                    }
                } else {
                    log.debug("MQTT client 接收的payload不是平台标准协议，topic = {} ，type = {} ，key = {}", topic, getType(), getKey());
                    receivedMessage = ReceivedMessage.builder(topic, payload)
                        .build();
                    receivedMessage.setNetworkType(getType());
                    receivedMessage = consumers.apply(receivedMessage);
                }

                if (receivedMessage != null && receivedMessage.getDeviceId() != null && !receivedMessage.getDeviceId().isEmpty()) {
                    DeviceStateManager deviceStateManager = DeviceStateManagerHolder.get();
                    if (deviceStateManager != null) {
                        deviceStateManager.reportAlive(receivedMessage.getDeviceId(), receivedMessage.getNetworkType());
                    }
                } else {
                    log.warn("设备状态信息管理因ReceivedMessage.deviceId不存在，忽略设备状态管理，ReceivedMessage = {}", receivedMessage == null ? "null" : receivedMessage.toString());
                }
            }

            @Override
            public void deliveryComplete(IMqttDeliveryToken token) {
                log.debug("MQTT client 消息投递完成：type = {} ，key = {}", getType(), getKey());
            }
        };
    }


    private static final List<String> PLATFORM_FIELDS = Arrays.asList(
        "kay", "messageId", "source", "payload", "headers", "route", "timestamp", "reply", "status", "error"
    );

    private boolean isPlatformMessageFormat(byte[] payload) {
        try {
            JsonNode node = objectMapper.readTree(payload);
            // 至少包含 4 个关键字段才认为是平台协议，避免误判普通 JSON
            int matchCount = 0;
            for (String field : PLATFORM_FIELDS) {
                if (node.has(field)) {
                    matchCount++;
                }
            }
            return matchCount >= 4;
        } catch (Exception e) {
            return false;
        }
    }

    private synchronized void changeNetworkState(NetworkStateEnum newState) {
        currentState = newState;
        NetworkStateManager networkStateManager = NetworkStateManagerHolder.get();

        if (networkStateManager != null) {
            try {
                networkStateManager.notifyStateChange(getKey(), newState);
            } catch (Exception e) {
                log.error("MQTT client 网络状态处理器发生异常：type = {} ，key = {}", getType(), getKey(), e);
            }
        }
    }


}
